package com.qupeng.demo.kafka.kafkaspringbootproducer.admin;

import com.qupeng.demo.kafka.kafkaspringbootproducer.admin.model.Topic;
import com.qupeng.demo.kafka.kafkaspringbootproducer.config.KafkaProperties;
import com.qupeng.demo.kafka.kafkaspringbootproducer.model.CommonResponse;
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.Uuid;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaAdmin;
import org.springframework.web.bind.annotation.*;

import java.util.Collections;
import java.util.concurrent.ExecutionException;

@RestController
@RequestMapping("/admin/async/topics")
public class TopicAsyncAdminController {

    private static final Logger LOGGER = LoggerFactory.getLogger(TopicAsyncAdminController.class);

    @Autowired
    private KafkaAdmin kafkaAdmin;

    @Autowired
    private AdminClient adminClient;

    @Autowired
    private KafkaProperties kafkaProperties;

    @PostMapping("")
    public CommonResponse<Topic> createTopic(@RequestBody Topic topic) {

        CreateTopicsResult newTopicResult = adminClient.createTopics(Collections.singletonList(new NewTopic(topic.getName(), topic.getPartitionCount(), topic.getReplicaFactor())));
        newTopicResult.config(topic.getName()).whenComplete((topicDsp, error) -> {
            if (null == error) {
                Uuid topicId = null;
                try {
                    topicId = newTopicResult.topicId(topic.getName()).get();
                    TopicDescription topicDescription = adminClient.describeTopics(Collections.singletonList(topic.getName())).topicIdValues().get(topicId).get();
                    LOGGER.info("Create topic successfully: {}", Util.convert(topicDescription));
                } catch (InterruptedException | ExecutionException e) {
                    LOGGER.error("Describe topic failed: {}", topic);
                }
            } else {
                LOGGER.error("Create topic failed: {}", topic, error);
            }
        });
        return new CommonResponse<>("kafka-admin", topic);
    }
}
